package com.example.grpc.client.helloworldStream;

import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.Status;
import io.grpc.StatusRuntimeException;
import io.grpc.examples.hellostreaming.HelloReply;
import io.grpc.examples.hellostreaming.HelloRequest;
import io.grpc.examples.hellostreaming.StreamingGreeterGrpc;
import io.grpc.stub.ClientCallStreamObserver;
import io.grpc.stub.ClientResponseObserver;
import io.grpc.stub.StreamObserver;
import lombok.SneakyThrows;

import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public class StreamSayHelloStreamClient {
    @SneakyThrows
    public static void main(String[] args) {
        final CountDownLatch done = new CountDownLatch(1);

        ManagedChannel channel = ManagedChannelBuilder.forAddress("172.16.36.76", 29080)
                .usePlaintext()
                .build();
        HelloRequest request = HelloRequest.newBuilder().setName("weiwei").build();
        StreamingGreeterGrpc.StreamingGreeterStub blockingStub = StreamingGreeterGrpc.newStub(channel);

        ClientResponseObserver<HelloRequest,HelloReply> clientResponseObserver =
                new ClientResponseObserver<HelloRequest, HelloReply>() {

            ClientCallStreamObserver<HelloRequest> requestStream;

            @Override
            public void beforeStart(ClientCallStreamObserver<HelloRequest> requestStream) {
                this.requestStream = requestStream;
                // Set up manual flow control for the response stream. It feels backwards to configure the response
                // stream's flow control using the request stream's observer, but this is the way it is.
                requestStream.disableAutoRequestWithInitial(1);

                // Set up a back-pressure-aware producer for the request stream. The onReadyHandler will be invoked
                // when the consuming side has enough buffer space to receive more messages.
                //
                // Messages are serialized into a transport-specific transmit buffer. Depending on the size of this buffer,
                // MANY messages may be buffered, however, they haven't yet been sent to the server. The server must call
                // request() to pull a buffered message from the client.
                //
                // Note: the onReadyHandler's invocation is serialized on the same thread pool as the incoming
                // StreamObserver's onNext(), onError(), and onComplete() handlers. Blocking the onReadyHandler will prevent
                // additional messages from being processed by the incoming StreamObserver. The onReadyHandler must return
                // in a timely manner or else message processing throughput will suffer.
                requestStream.setOnReadyHandler(new Runnable() {
                    // An iterator is used so we can pause and resume iteration of the request data.
                    Iterator<String> iterator = names().iterator();

                    @Override
                    public void run() {
                        // Start generating values from where we left off on a non-gRPC thread.
                        while (requestStream.isReady()) {
                            if (iterator.hasNext()) {
                                // Send more messages if there are more messages to send.
                                String name = iterator.next();
                                System.out.println("--> " + name);
                                HelloRequest request = HelloRequest.newBuilder().setName(name).build();
                                requestStream.onNext(request);
                            } else {
                                // Signal completion if there is nothing left to send.
                                requestStream.onCompleted();
                            }
                        }
                    }
                });
            }

            @Override
            public void onNext(HelloReply helloReply) {
                System.out.println("<-- " + helloReply.getMessage());
                // Signal the sender to send one message.
                requestStream.request(1);
            }

            @Override
            public void onError(Throwable throwable) {
                throwable.printStackTrace();
                done.countDown();
            }

            @Override
            public void onCompleted() {
                System.out.println("all done");
                done.countDown();
            }
        };

        /*StreamObserver<HelloRequest> requestObserver = blockingStub.sayHelloStreaming(new StreamObserver<HelloReply>() {
            @Override
            public void onNext(HelloReply helloReply) {
                System.out.println("获取到消息："+helloReply.getMessage());
//                result.add(helloReply.getMessage());
            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println("sayHelloStream Failed : "+ Status.fromThrowable(throwable));
                done.countDown();
            }

            @Override
            public void onCompleted() {
                System.out.println("Finish sayHelloStream");
                done.countDown();
            }
        });

        for (int i=1;i<11;i++){
            String name = "weiwei -- " + i;
            System.out.println("发送 "+ "weiwei" + "--"+i);
            HelloRequest req = HelloRequest.newBuilder().setName(name).build();
            requestObserver.onNext(req);
        }*/

        blockingStub.sayHelloStreaming(clientResponseObserver);

        done.await();
        channel.shutdown();
        channel.awaitTermination(1, TimeUnit.SECONDS);

    }


    private static List<String> names() {
        return Arrays.asList(
                "Sophia",
                "Jackson",
                "Emma",
                "Aiden",
                "Olivia");
    }
}
